package org.example.rocketmq_product.controller;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

@RestController
@RequestMapping("/product")
public class ProductController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /*topic 需要先在rocketmq 上手动新建 然后才能使用，而不是通过代码新建 topic*/
    private final String TOPIC = "linuxGroup1";

    private final String TAG = "tag1";//这个一般就是用于过滤

    /*发送单向消息
指发送消息后，不需要等待Broker的响应，直接返回。这种方式适用于不需要关注消息发送结果的场景，如日志记录、统计信息等。*/
    @GetMapping("/sendOneWay")
    public String sendOneWay() {
        Message<String> helloRocketmqOneWay =
                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq one way"))
                        .setHeader("KEYS", UUID.randomUUID().toString())//设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
                        .build();

        //给消息打上射手的标签。主题+tag，中间用“:”分隔,主要是用于消息的过滤，比如说在消费的时候，只消费ESS标签下的消息
        rocketMQTemplate.sendOneWay(TOPIC , helloRocketmqOneWay);
        return "sendOneWay success";
    }

    /*同步发送消息*/
    /*syncSend方法会阻塞当前线程，直到消息发送完成并收到了消息服务器的响应。如果消息发送成功，
    syncSend方法会返回一个SendResult对象，包含了消息的发送状态、消息ID等信息。如果消息发送失败，
    syncSend方法会抛出一个MessagingException异常。*/
    @GetMapping("/syncSend")
    public String syncSend() {
        Message<String> helloRocketmqSync =
                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq sync"))
                        .setHeader("KEYS", UUID.randomUUID().toString())
                        .build();

        rocketMQTemplate.syncSend(TOPIC, helloRocketmqSync);
        return "syncSend success";
    }

//    异步发送消息
//    asyncSend方法不会阻塞当前线程，而是在另一个线程中异步发送消息。因此，asyncSend方法会立即返回，
//    不会等待消息发送完成。如果需要等待消息发送完成并处理发送结果，可以使用SendCallback回调接口。
    @GetMapping("/asyncSend")
    public String asyncSend() {
        Message<String> helloRocketmqAsync =
                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq async"))
                        .setHeader("KEYS", UUID.randomUUID().toString())
                        .build();

        rocketMQTemplate.asyncSend(TOPIC, helloRocketmqAsync, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("send success");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("send fail");
            }
        });
        return "asyncSend success";
    }

//    顺序消息
//顺序消息是 Apache RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，
// 从而实现业务场景中的顺序处理。 相比其他类型消息，顺序消息在发送、存储和投递的处理过程中，更多强调多条消息间的先后顺序关系。

//    单向顺序消息
    @GetMapping("/sendOneWayOrderly")
    public String sendOneWayOrderly() {
        Message<String> helloRocketmqOneWayOrderly =
                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq one way orderly"))
                        .setHeader("KEYS", UUID.randomUUID().toString())
                        .build();

        rocketMQTemplate.sendOneWayOrderly(TOPIC, helloRocketmqOneWayOrderly, "1");
        //hashkey 用于保证消息的有序性，同一个hashkey的消息会被发送到同一个队列中  一般都是数据的某个字段的值
        //举个例子，一个货物从下单到送达之间经历很多步骤 下单生成预支付订单--客户付款--减少库存--发货--收获，
        // 这几个步骤都有着相同的订单号，所以我们使用订单号作为hashkey，这样就能保证同一个订单号的消息会被发送到同一个队列中
        return "sendOneWayOrderly success";
    }

//    异步发送顺序消息
    @GetMapping("/asyncSendOrderly")
    public String asyncSendOrderly() {
        Message<String> helloRocketmqAsyncOrderly =
                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq async orderly"))
                        .setHeader("KEYS", UUID.randomUUID().toString())
                        .build();

        rocketMQTemplate.asyncSendOrderly(TOPIC, helloRocketmqAsyncOrderly, "1", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("send success");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("send fail");
            }
        });
        return "asyncSendOrderly success";
    }

//    定时/延时消息
//    一种是  10s后才能消费到消息  一种是 指定时间 才能消费到消息
    //10s后才能消费到消息
//    @GetMapping("/sendDelay")
//    public String sendDelay() {
//        Message<String> helloRocketmqDelay =
//                MessageBuilder.withPayload(JSON.toJSONString("hello rocketmq delay"))
//                        .setHeader("KEYS", UUID.randomUUID().toString())
//                        .build();
//
//        rocketMQTemplate.syncSend(TOPIC + ":delay", helloRocketmqDelay, 10000L, 3);
//        return "sendDelay success";
//    }


}
